package lyw.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import lyw.app.BaseAppV1;
import lyw.bean.IsNewOrActivityBean;
import lyw.common.Constant;
import lyw.util.AtguiguUtil;
import lyw.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author: lyw
 * @Date: 2022/9/1 16:37
 * @Describe:
 */
public class Dws_01_DwsIsNewOrActivity extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_01_DwsIsNewOrActivity().init(
                4001,
                2,
                "Dws_01_DwsIsNewOrActivity",
                Constant.TOPIC_DWD_TRAFFIC_UV_DETAIL
        );
    }

    /**
     * 浏览记录中有uid的都是已经登陆的用户
     * 使用状态记录用户的上一次登陆时间，若状态为空，则是新增用户；
     * 若不为空，说明是老用户
     *
     * @return
     */

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {

        SingleOutputStreamOperator<IsNewOrActivityBean> beanStream = findNewUserOrAct(stream);

        SingleOutputStreamOperator<IsNewOrActivityBean> resultStream = windowAndAgg(beanStream);
        writeToDoris(resultStream);
    }

    private void writeToDoris(SingleOutputStreamOperator<IsNewOrActivityBean> resultStream) {
        resultStream
                .map(bean -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
                    return JSON.toJSONString(bean, config);
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_is_new_or_activity"));
    }

    private SingleOutputStreamOperator<IsNewOrActivityBean> windowAndAgg(
            SingleOutputStreamOperator<IsNewOrActivityBean> beanStream) {

        return beanStream.map(new MapFunction<IsNewOrActivityBean, IsNewOrActivityBean>() {
            @Override
            public IsNewOrActivityBean map(IsNewOrActivityBean value) throws Exception {
                System.out.println(value.getTs());
                System.out.println(value+"value");
                return value;
            }
        })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<IsNewOrActivityBean>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<IsNewOrActivityBean>() {
                            @Override
                            public IsNewOrActivityBean reduce(IsNewOrActivityBean bean1,
                                                              IsNewOrActivityBean bean2) throws Exception {
                                System.out.println("bean1:" + bean1 + " | " + "bean2:" + bean2);
                                bean1.setActCt(bean1.getActCt() + bean2.getActCt());
                                bean1.setIsNewCt(bean1.getIsNewCt() + bean2.getIsNewCt());
                                System.out.println("------------bean1: " + bean1);
                                return bean1;
                            }
                        },
                        new ProcessAllWindowFunction<IsNewOrActivityBean, IsNewOrActivityBean, TimeWindow>() {
                            @Override
                            public void process(Context ctx,
                                                Iterable<IsNewOrActivityBean> elements,
                                                Collector<IsNewOrActivityBean> out) throws Exception {
                                System.out.println("------------------------------" );
                                IsNewOrActivityBean bean = elements.iterator().next();

                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));

                                bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis()));

                                out.collect(bean);

                            }
                        }
                );
    }

    private SingleOutputStreamOperator<IsNewOrActivityBean> findNewUserOrAct(
            DataStreamSource<String> stream){

        return stream
                .map(JSONObject::parseObject)
                .filter(obj -> obj.getJSONObject("common").getString("uid") != null)
                .keyBy(obj -> obj.getJSONObject("common").getString("uid"))
                .process(new KeyedProcessFunction<String, JSONObject, IsNewOrActivityBean>() {

                    private ValueState<String> lastLoginDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastLoginDateState =
                                getRuntimeContext().getState(new ValueStateDescriptor<String>("lastLoginDateState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj,
                                               Context ctx,
                                               Collector<IsNewOrActivityBean> out) throws Exception {

                        Long ts = obj.getLong("ts");

                        String thisLoginDate = AtguiguUtil.toDate(ts);
                        // 最后一次登录日期
                        String lastLoginDate = lastLoginDateState.value();

                        Long actCt = 0L;

                        Long isNewCt = 0L;
                        // 如果这次登录日期与最后一次登录日期不相等, 则这次是今天的第一次登录
                        if (!thisLoginDate.equals(lastLoginDate)) {
                            // 今天第一次登录
                            lastLoginDateState.update(thisLoginDate);  // 把状态更新为这次登录到日期
                            actCt = 1L;

                            //todo 状态为空，则是新用户
                            if (lastLoginDate == null) {

                                isNewCt = 1L;
                            }
                        }

                        if (actCt == 1) {
                            out.collect(new IsNewOrActivityBean("", "",
                                    "",
                                    actCt,
                                    isNewCt,
                                    ts
                            ));

                        }
                    }
                });

    }
}
